package com.duowan.realtime.dao.impl;

import com.duowan.realtime.dao.BaseSpringJdbcDao;
import com.duowan.realtime.dao.SqlJobDao;
import com.duowan.realtime.model.SqlJob;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.namedparam.BeanPropertySqlParameterSource;

import java.util.List;

/**
 * @version 1.0
 * @author: tangsicheng
 * @date: 14-8-27
 * @since 1.0
 */
public class SqlJobDaoImpl extends BaseSpringJdbcDao implements SqlJobDao {
    private static final Logger logger = LoggerFactory.getLogger(SqlJobDaoImpl.class);

    private RowMapper<SqlJob> entityRowMapper = new BeanPropertyRowMapper<SqlJob>(getEntityClass());

    public RowMapper<SqlJob> getEntityRowMapper() {
        return entityRowMapper;
    }


    @Override
    public Class<SqlJob> getEntityClass() {
        return SqlJob.class;
    }


    @Override
    public void insert(SqlJob entity) {
        String sql = "insert into sql_job "
                + " (name,query_statement,into_redis,into_db,error,passport) "
                + " values "
                + " (:name,:queryStatement,:intoRedis,:intoDb,:errorMessage,:passport)";

        getNamedParameterJdbcTemplate().update(sql,new BeanPropertySqlParameterSource(entity));
    }

    @Override
    public int deleteById(String sqlId) {
        String sql = "delete from sql_job where  sql_id = ? ";
        return  getSimpleJdbcTemplate().update(sql,  sqlId);
    }



    @Override
    public SqlJob getSqlJobById(String sqlId) {
        String sql = "SELECT sql_id AS sqlId, name AS name, into_redis AS intoRedis, query_statement AS queryStatement," +
                "into_db AS intoDb,error AS errorMessage,passport AS passport FROM sql_job where sql_id = ?";
        return (SqlJob) DataAccessUtils.singleResult(getSimpleJdbcTemplate().query(sql, getEntityRowMapper(), sqlId));
    }

    @Override
    public List<SqlJob> getSqlJobs() {
        JdbcTemplate jdbcTemplate = this.getJdbcTemplate();
        String sql  =  "SELECT sql_id AS sqlId,name AS name, query_statement AS queryStatement, into_redis AS intoRedis, into_db AS intoDb, error AS errorMessage FROM sql_job";
        List<SqlJob> jobList = jdbcTemplate.query(sql, getEntityRowMapper());
        return jobList;
    }

    @Override
    public void updateJobError(String sqlIds, String errorMessage) {
        String sql = "UPDATE sql_job SET error=? where  sql_id =? ";
        getSimpleJdbcTemplate().update(sql, errorMessage, sqlIds);

    }

    @Override
    public SqlJob getSqlJobByQueryStatement(String queryStatement) {

        String sql = "SELECT name AS name, into_redis AS intoRedis, query_statement AS queryStatement," +
                "into_db AS intoDb,error AS errorMessage,passport AS passport FROM sql_job where query_statement = ?";

        List<SqlJob> columnList = getJdbcTemplate().query(sql, getEntityRowMapper(), queryStatement);

        if(CollectionUtils.isEmpty(columnList)){
            return null;
        }else{
            return columnList.get(0);
        }
    }


	@Override
	public int update(SqlJob entity) {
		String sql = "update sql_job set "
				+ " name=:name,into_redis=:intoRedis,query_statement=:queryStatement "
				+ " where  sql_id = :sqlId ";
		return getNamedParameterJdbcTemplate().update(sql, new BeanPropertySqlParameterSource(entity));
	}
}
